package com.wudl.flink.test;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Iterator;

/**
 * @author wudl
 * @date 2021-11-02 00:02:17
 * @description
 */
public class FlinkPOJO {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);
        DataStream<String> source =
                env.addSource(new SourceFunction<String>() {
                    @Override
                    public void run(SourceContext<String> sourceContext) throws Exception {
                        while (true) {
                            sourceContext.collect("{\"device\":\"rand-numeric\",\"id\":\"b4728895-741f-466a-b87b-79c7590893b4\",\"origin\":\"1591095418904441036\",\"readings\":[{\"origin\":\"1591095418904328442\",\"valueType\":\"Int64\",\"name\":\"int\",\"device\":\"rand-numeric\",\"value\":\"0\"}]}");
                            Thread.sleep(1000);
                        }
                    }

                    @Override
                    public void cancel() {

                    }
                });
        DataStream<Info> parsedSource =
                source.map(new MapFunction<String, Info>() {
                    @Override
                    public Info map(String s) throws Exception {
                        JSONObject gson = new JSONObject();
                        return gson.getObject(s, Info.class);
                    }
                });

        DataStream<String> output = parsedSource.keyBy(Info::getId).timeWindow(Time.seconds(1))
                .process(new ProcessWindowFunction<Info, String, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<Info> iterable, Collector<String> collector) throws Exception {
                        int count = 0;
                        Iterator<Info> iterator = iterable.iterator();
                        while (iterator.hasNext()) {
                            count++;
                            iterator.next();
                        }
                        collector.collect(String.format("key : %s, size : %s", s, count));
                    }
                });
        output.print();
        env.execute();
    }


    public class Info {
        public String getDevice() {
            return device;
        }

        public void setDevice(String device) {
            this.device = device;
        }

        public String getId() {
            return id;
        }

        public void setId(String id) {
            this.id = id;
        }

        public String getOrigin() {
            return origin;
        }

        public void setOrigin(String origin) {
            this.origin = origin;
        }

        public Reading[] getReadings() {
            return readings;
        }

        public void setReadings(Reading[] readings) {
            this.readings = readings;
        }

        public String device;
        public String id;
        public String origin;
        public Reading[] readings;

        public Info() {

        }


    }

    public class Reading {
        public String origin;
        public String valueType;
        public String name;
        public String device;
        public String value;

        public Reading() {

        }
    }
}
